//欧盘大数据文件加载 From HDFS
package caiqr.utils

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SQLContext}
import org.apache.spark.rdd.RDD

object AllEuropeInputFile {


  // 加载 欧赔大文件, 返回 DF.
  def load(sc: SparkContext, sqlContext: SQLContext, filename: String): DataFrame ={

    val people = sc.textFile(filename)
    val schemaString = "match_id,company_id,match_time,season_id,match_desc,season_pre,group_pre,host_id,away_id,home_match_result,away_match_result,score,init_win,init_draw,init_loss,curr_win,curr_draw,curr_loss,myear,mmonth,init_win_range,init_draw_range,init_loss_range,curr_win_range,curr_draw_range,curr_loss_range,win_diff,draw_diff,loss_diff,src_init_ret,src_curr_ret,init_ret,curr_ret,src_init_win,src_init_draw,src_init_loss,src_curr_win,src_curr_draw,src_curr_loss,src_init_win_ratio,src_init_draw_ratio,src_init_loss_ratio,init_win_ratio,init_win_ratio,init_loss_ratio,src_curr_win_ratio,src_curr_draw_ratio,src_curr_loss_ratio,curr_win_ratio,curr_draw_ratio,curr_loss_ratio,goal"
    import org.apache.spark.sql.Row
    import org.apache.spark.sql.types.{StringType, StructField, StructType}
    val schema = StructType(schemaString.split(",").map(fieldName => StructField(fieldName, StringType, nullable = true)))

    //加工赔率值
    val rowRDD = people.map(_.split("\t")).map{p =>

      val init_win = p(12).toInt
      val init_draw = p(13).toInt
      val init_loss = p(14).toInt
      val curr_win = p(15).toInt
      val curr_draw = p(16).toInt
      val curr_loss = p(17).toInt

      //1).赔率范围
      val init_win_range = (init_win/100)*100
      val init_draw_range = (init_draw/100)*100
      val init_loss_range = (init_loss/100)*100
      val curr_win_range = (curr_win/100)*100
      val curr_draw_range = (curr_draw/100)*100
      val curr_loss_range = (curr_loss/100)*100

      //2).赔率变化比例,  (终盘-初盘)/初盘=比例
      val win_diff = ((curr_win-init_win).toDouble/init_win*100).round.toInt
      val draw_diff = ((curr_draw-init_draw).toDouble/init_draw*100).round.toInt
      val loss_diff = ((curr_loss-init_loss).toDouble/init_loss*100).round.toInt

      //3).返奖率 (1/(1/w+1/d+1/l))
      val src_init_ret = (1.0/(1.0/init_win+1.0/init_draw+1.0/init_loss)).round.toInt // 8902
    val src_curr_ret = (1.0/(1.0/curr_win+1.0/curr_draw+1.0/curr_loss)).round.toInt // 8902
    val init_ret = (1.0/(1.0/init_win+1.0/init_draw+1.0/init_loss)/1000*100).round.toInt //89
    val curr_ret = (1.0/(1.0/curr_win+1.0/curr_draw+1.0/curr_loss)/1000*100).round.toInt //89
    val all_init_ret = (1.0/(1.0/init_win+1.0/init_draw+1.0/init_loss)/1000) //0.89022323232
    val all_curr_ret = (1.0/(1.0/curr_win+1.0/curr_draw+1.0/curr_loss)/1000)


      //4).标准赔率=赔率/返奖率
      val src_init_win = (init_win/all_init_ret).round.toInt
      val src_init_draw = (init_draw/all_init_ret).round.toInt
      val src_init_loss = (init_loss/all_init_ret).round.toInt
      val all_init_win = init_win/all_init_ret
      val all_init_draw = init_draw/all_init_ret
      val all_init_loss = init_loss/all_init_ret

      val src_curr_win = (curr_win/all_curr_ret).round.toInt
      val src_curr_draw = (curr_draw/all_curr_ret).round.toInt
      val src_curr_loss = (curr_loss/all_curr_ret).round.toInt
      val all_curr_win = curr_win/all_curr_ret
      val all_curr_draw = curr_draw/all_curr_ret
      val all_curr_loss = curr_loss/all_curr_ret


      //5).根据标准赔率,计算标准概率比例(和为100%)
      //标准概率=1/标准赔率
      val src_init_win_ratio = (1/all_init_win*1000*100).round.toInt    //74
      val src_init_draw_ratio = (1/all_init_draw*1000*100).round.toInt  //15
      val src_init_loss_ratio = (1/all_init_loss*1000*100).round.toInt  //11
      val init_win_ratio = src_init_win_ratio - src_init_win_ratio%5    //70,取整(以5倍数)
      val init_draw_ratio = src_init_draw_ratio - src_init_draw_ratio%5 //15
      val init_loss_ratio = src_init_loss_ratio - src_init_loss_ratio%5 //10

      val src_curr_win_ratio = (1/all_curr_win*1000*100).round.toInt    //74
      val src_curr_draw_ratio = (1/all_curr_draw*1000*100).round.toInt  //15
      val src_curr_loss_ratio = (1/all_curr_loss*1000*100).round.toInt  //11
      val curr_win_ratio = src_curr_win_ratio - src_curr_win_ratio%5    //70
      val curr_draw_ratio = src_curr_draw_ratio - src_curr_draw_ratio%5 //15
      val curr_loss_ratio = src_curr_loss_ratio - src_curr_loss_ratio%5 //10

      Row(p(0), p(1), p(2), p(3), p(4), p(5), p(6), p(7), p(8), p(9), p(10), p(11), p(12), p(13), p(14), p(15), p(16), p(17), p(18), p(19), init_win_range.toString,init_draw_range.toString,init_loss_range.toString,curr_win_range.toString,curr_draw_range.toString,curr_loss_range.toString,win_diff.toString,draw_diff.toString,loss_diff.toString,src_init_ret.toString,src_curr_ret.toString,init_ret.toString,curr_ret.toString,src_init_win.toString,src_init_draw.toString,src_init_loss.toString,src_curr_win.toString,src_curr_draw.toString,src_curr_loss.toString,src_init_win_ratio.toString,src_init_draw_ratio.toString,src_init_loss_ratio.toString,init_win_ratio.toString,init_draw_ratio.toString,init_loss_ratio.toString,src_curr_win_ratio.toString,src_curr_draw_ratio.toString,src_curr_loss_ratio.toString,curr_win_ratio.toString,curr_draw_ratio.toString,curr_loss_ratio.toString, p(20))
    }


    val europe_df = sqlContext.createDataFrame(rowRDD, schema)
    europe_df

  }



  // 加载需要计算的比赛所有盘口信息文件
  def load_match_file(sc: SparkContext, sqlContext: SQLContext, filename: String): DataFrame ={
    val people = sc.textFile(filename)
    val schemaString = "match_id,company_id,match_time,season_id,match_desc,season_pre,group_pre,host_id,away_id,init_win,init_draw,init_loss,curr_win,curr_draw,curr_loss,myear,mmonth,init_win_range,init_draw_range,init_loss_range,curr_win_range,curr_draw_range,curr_loss_range,win_diff,draw_diff,loss_diff,src_init_ret,src_curr_ret,init_ret,curr_ret,src_init_win,src_init_draw,src_init_loss,src_curr_win,src_curr_draw,src_curr_loss,src_init_win_ratio,src_init_draw_ratio,src_init_loss_ratio,init_win_ratio,init_win_ratio,init_loss_ratio,src_curr_win_ratio,src_curr_draw_ratio,src_curr_loss_ratio,curr_win_ratio,curr_draw_ratio,curr_loss_ratio"
    import org.apache.spark.sql.Row
    import org.apache.spark.sql.types.{StringType, StructField, StructType}
    val schema =
      StructType(
        schemaString.split(",").map(fieldName => StructField(fieldName, StringType, nullable = true)))

  //加工赔率值
  val rowRDD = people.map(_.split("\t")).map{p =>

    val init_win = p(9).toInt
    val init_draw = p(10).toInt
    val init_loss = p(11).toInt
    val curr_win = p(12).toInt
    val curr_draw = p(13).toInt
    val curr_loss = p(14).toInt

    //////// 标准赔率
    //1).赔率范围
    val init_win_range = (init_win/100)*100
    val init_draw_range = (init_draw/100)*100
    val init_loss_range = (init_loss/100)*100
    val curr_win_range = (curr_win/100)*100
    val curr_draw_range = (curr_draw/100)*100
    val curr_loss_range = (curr_loss/100)*100

    //2).赔率变化比例,  (终盘-初盘)/初盘=比例
    val win_diff = ((curr_win-init_win).toDouble/init_win*100).round.toInt
    val draw_diff = ((curr_draw-init_draw).toDouble/init_draw*100).round.toInt
    val loss_diff = ((curr_loss-init_loss).toDouble/init_loss*100).round.toInt

    //3).返奖率 (1/(1/w+1/d+1/l))
    val src_init_ret = (1.0/(1.0/init_win+1.0/init_draw+1.0/init_loss)).round.toInt // 8902
    val src_curr_ret = (1.0/(1.0/curr_win+1.0/curr_draw+1.0/curr_loss)).round.toInt // 8902
    val init_ret = (1.0/(1.0/init_win+1.0/init_draw+1.0/init_loss)/1000*100).round.toInt //89
    val curr_ret = (1.0/(1.0/curr_win+1.0/curr_draw+1.0/curr_loss)/1000*100).round.toInt //89
    val all_init_ret = (1.0/(1.0/init_win+1.0/init_draw+1.0/init_loss)/1000) //0.89022323232
    val all_curr_ret = (1.0/(1.0/curr_win+1.0/curr_draw+1.0/curr_loss)/1000)


    //4).标准赔率=赔率/返奖率
    val src_init_win = (init_win/all_init_ret).round.toInt
    val src_init_draw = (init_draw/all_init_ret).round.toInt
    val src_init_loss = (init_loss/all_init_ret).round.toInt
    val all_init_win = init_win/all_init_ret
    val all_init_draw = init_draw/all_init_ret
    val all_init_loss = init_loss/all_init_ret

    val src_curr_win = (curr_win/all_curr_ret).round.toInt
    val src_curr_draw = (curr_draw/all_curr_ret).round.toInt
    val src_curr_loss = (curr_loss/all_curr_ret).round.toInt
    val all_curr_win = curr_win/all_curr_ret
    val all_curr_draw = curr_draw/all_curr_ret
    val all_curr_loss = curr_loss/all_curr_ret


    //5).根据标准赔率,计算标准概率比例(和为100%)
    //标准概率=1/标准赔率
    val src_init_win_ratio = (1/all_init_win*1000*100).round.toInt    //74
    val src_init_draw_ratio = (1/all_init_draw*1000*100).round.toInt  //15
    val src_init_loss_ratio = (1/all_init_loss*1000*100).round.toInt  //11
    val init_win_ratio = src_init_win_ratio - src_init_win_ratio%5    //70,取整(以5倍数)
    val init_draw_ratio = src_init_draw_ratio - src_init_draw_ratio%5 //15
    val init_loss_ratio = src_init_loss_ratio - src_init_loss_ratio%5 //10

    val src_curr_win_ratio = (1/all_curr_win*1000*100).round.toInt    //74
    val src_curr_draw_ratio = (1/all_curr_draw*1000*100).round.toInt  //15
    val src_curr_loss_ratio = (1/all_curr_loss*1000*100).round.toInt  //11
    val curr_win_ratio = src_curr_win_ratio - src_curr_win_ratio%5    //70
    val curr_draw_ratio = src_curr_draw_ratio - src_curr_draw_ratio%5 //15
    val curr_loss_ratio = src_curr_loss_ratio - src_curr_loss_ratio%5 //10


    Row(p(0), p(1), p(2), p(3), p(4), p(5), p(6), p(7), p(8), p(9), p(10), p(11), p(12), p(13), p(14), p(15), p(16), init_win_range.toString,init_draw_range.toString,init_loss_range.toString,curr_win_range.toString,curr_draw_range.toString,curr_loss_range.toString,win_diff.toString,draw_diff.toString,loss_diff.toString,src_init_ret.toString,src_curr_ret.toString,init_ret.toString,curr_ret.toString,src_init_win.toString,src_init_draw.toString,src_init_loss.toString,src_curr_win.toString,src_curr_draw.toString,src_curr_loss.toString,src_init_win_ratio.toString,src_init_draw_ratio.toString,src_init_loss_ratio.toString,init_win_ratio.toString,init_draw_ratio.toString,init_loss_ratio.toString,src_curr_win_ratio.toString,src_curr_draw_ratio.toString,src_curr_loss_ratio.toString,curr_win_ratio.toString,curr_draw_ratio.toString,curr_loss_ratio.toString)
  }


  val europe_df = sqlContext.createDataFrame(rowRDD, schema)
  europe_df
  }
//



}
